package com.erbao.disruptor.lambda;

import com.erbao.disruptor.LongEvent;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;

/**
 * @author CaoBaoYin
 * @since 2022/3/10 7:37
 */
public class MainLambda {

    public static void main(String[] args) {
        // 创建工厂
        // 队列大小
        int ringBufferSize = 1024;
        // 创建disruptor。线程工厂里创建线程，在这个线程里调用消费者，不同的消费者会在不同的线程
        Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(LongEvent::new, ringBufferSize, DaemonThreadFactory.INSTANCE);
        // 指定消费者
        disruptor.handleEventsWith((event, sequence, endOfBatch) -> {
            Thread thread = Thread.currentThread();
            System.out.println(thread.getThreadGroup()+":"+thread.getName()+":消费序号"+sequence+"，对象值："+event.getValue());
        });

        // 启动
        disruptor.start();

        // 生产event
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        // 不带参数
        ringBuffer.publishEvent((event, sequence) -> {event.setValue(1000L);});

        // 带一个参数
        ringBuffer.publishEvent((event, sequence, arg0) -> {event.setValue(arg0);}, 1002L);

        // 带两个参数
        ringBuffer.publishEvent((LongEvent event, long sequence, Long arg0, Long arg1) -> {event.setValue(arg0+arg1);}, 1002L, 1002L);

        // 带三个参数
        ringBuffer.publishEvent(( event,  sequence,  arg0,  arg1,  arg2) -> {
            event.setValue(arg0+arg1+arg2);
        }, 1003L, 1003L, 1003L);

        // 带不定参数
        ringBuffer.publishEvent(( event,  sequence,  objects) -> {
            long result = 0L;

            for (Object arg : objects) {
                long l = 0L;
                if (arg instanceof Long) {
                    l = (long) arg;
                }
                result += l;
            }

            event.setValue(result);
        }, 1006L, 1006L, 1006L, 1006L, 1006L, 1006L);

        // 主线程会自动结束

        /*try {
            disruptor.shutdown(5L, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            e.printStackTrace();
        }*/

        /*try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }*/
    }

}
